package com.atguigu.realtime.clients;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.realtime.constants.PrefixConstant;
import com.atguigu.realtime.constants.TopicConstant;
import com.atguigu.realtime.utils.JedisUtil;
import com.atguigu.realtime.utils.KafkaProducerUtil;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import redis.clients.jedis.Jedis;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;

/**
 * Created by Smexy on 2022/8/27
 *
 *   采集事实数据:
 *          order_info : insert
 *          order_detail : insert
 *
 *   采集维度数据:
 *          user_info :   insert | update
 *
 *          设计在redis中存储的
 *              粒度：   一个用户是一个K-V
 *                  K:  user_id
 *                  V:  json
 *
 *              粒度：  所有用户是一个K-V
 *                  K:
 *                  V:  set
 *                      Hash[user_id, json]
 *
 *         后续的使用场景是:
 *              根据一笔订单的user_id 查询 用户的信息，最适合的就是 一个用户是一个K-V,效率最高。
 */
public class SaleDetailClient
{

    //程序是24h不停拉取数据，生产到kafka,jedis获取之后，无需关闭
    public static Jedis jedis = JedisUtil.getJedis();

    public static void main(String[] args) throws InvalidProtocolBufferException {

        //①先创建一个客户端对象CanalConnector
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop103", 11111), "example", null
            , null);

        //②使用客户端对象连接 Canal server端
        canalConnector.connect();

        // ③订阅表
        // 格式： 库名.表名
        canalConnector.subscribe("220409.*");


        //拉取数据
        while(true){

            Message message = canalConnector.get(100);

            if (message.getId() == -1){

                //当前没有拉取到数据
                System.out.println("当前没有新数据了，歇5s....");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                continue;

            }

            //说明拉到数据，解析数据
            //System.out.println(message);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {

                //sql写操作的表名
                String tableName = entry.getHeader().getTableName();

                //写入的数据，需要反序列化再使用
                ByteString storeValue = entry.getStoreValue();

                // 如果你只处理 insert,update,delete，它们的 EntryType 都是 ROWDATA
                CanalEntry.EntryType entryType = entry.getEntryType();

                if (entryType.equals(CanalEntry.EntryType.ROWDATA)){
                    parseData(storeValue,tableName);
                }


            }


        }



    }

    // 当前只奥 orderInfo这个表的 insert操作
    private static void parseData(ByteString storeValue,String tableName) throws InvalidProtocolBufferException {


        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

        //EventType : 写操作的关键字   insert ,update等
        if ("order_info".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            sendData(rowChange,true, TopicConstant.ORDER_INFO);
        }else if("order_detail".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            sendData(rowChange,true, TopicConstant.ORDER_DETAIL);
        }else if("user_info".equals(tableName) &&
                        (
                            rowChange.getEventType().equals(CanalEntry.EventType.INSERT)
                                ||
                            rowChange.getEventType().equals(CanalEntry.EventType.UPDATE)
                        )
               ){

            sendData(rowChange,false, null);
        }

    }

    /*
            提供一个标记，是否写入Kafka还是redis  TopicConstant.ORDER_INFO
     */
    private static void sendData(CanalEntry.RowChange rowChange,boolean ifSendToKafka,String topic){

        //这条insert语句，造成的多行的写操作变化
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

        //一个rowData写入的一行
        for (CanalEntry.RowData rowData : rowDatasList) {

            //把一行封装为一个JSON
            JSONObject jsonObject = new JSONObject();

            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

            for (CanalEntry.Column column : afterColumnsList) {

                jsonObject.put( column.getName(),column.getValue());

            }


            //发送数据
            if (ifSendToKafka){

                //模拟延迟
                /*int delayNum = new Random().nextInt(5);

                try {
                    Thread.sleep(delayNum * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/

                KafkaProducerUtil.sendData(jsonObject.toString(),topic );
            }else{
                jedis.set(PrefixConstant.user_info_redis_preffix + jsonObject.getString("id"), jsonObject.toString());
            }

        }

    }
}
